package com.atguigu.day07;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

//redis写入
public class Example03 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.addSource(new com.atguigu.day02.Example3.MySource())
                .addSink(
                        new RedisSink<com.atguigu.day02.Example3.Event>(
                                new FlinkJedisPoolConfig.Builder().setHost("hadoop102").build(), new RedisMapper<com.atguigu.day02.Example3.Event>() {
                            @Override
                            public RedisCommandDescription getCommandDescription() {
                                return new RedisCommandDescription(RedisCommand.HSET, "clicks");
                            }

                            @Override
                            public String getKeyFromData(com.atguigu.day02.Example3.Event event) {
                                return event.name;
                            }

                            @Override
                            public String getValueFromData(com.atguigu.day02.Example3.Event event) {
                                return event.url;
                            }
                        }
                        )
                );

        env.execute();
    }
}
